package com.atbeijing.gmall.realtime.utils;


import com.atbeijing.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/**
 * kafka工具类
 */
public class MyKafkaUtil {

    /**
     * 获取kafka消费者
     * @param topic 主题
     * @param groupId 消费者组名字
     * @return
     */
    public static FlinkKafkaConsumer getKafkaSource(String topic,String groupId){
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,GmallConfig.KAFKA_SERVER);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),props);
        kafkaSource.setStartFromEarliest();
        return kafkaSource;
    }

    /**
     * 获取flink封装的kafka生产者对象,一个流中的数据写到一个主题
     * @param topic 主题
     * @return
     */
    public static FlinkKafkaProducer<String> getKafkaSink(String topic){
        return new FlinkKafkaProducer<String>(GmallConfig.KAFKA_SERVER,topic,new SimpleStringSchema());
    }


    /**
     * 获取flink封装的kafka生产者对象,一个流中的数据要写到多个主题
     * @param kafkaSerializationSchema kafka序列化模型
     * @param <T>
     * @return
     */
    public static <T>FlinkKafkaProducer<T> getKafkaSinkBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema){
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, GmallConfig.KAFKA_SERVER);
        //设置事务超时时间
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 +"");
        //默认输出到哪个主题,序列化方式(向哪个主题写什么数据),生产者配置信息,精确一次性保证,事务保证
        return new FlinkKafkaProducer<T>(GmallConfig.DEFAULT_TOPIC,kafkaSerializationSchema,props,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    //拼接Kafka相关属性到DDL
    public static String getKafkaDDL(String topic,String groupId){
        String ddl="'connector' = 'kafka', " +
                " 'topic' = '"+topic+"',"   +
                " 'properties.bootstrap.servers' = '"+ GmallConfig.KAFKA_SERVER +"', " +
                " 'properties.group.id' = '"+groupId+ "', " +
                "  'format' = 'json', " +
                "  'scan.startup.mode' = 'latest-offset'  ";
        return  ddl;
    }
}
